package com.it.sink;


import com.it.operator.utils.SourceUtils;
import com.it.pojo.Event;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

/**
 * 将数据写入到redis中
 *
 * @author code1997
 */
public class RedisSinkDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        SingleOutputStreamOperator<Event> eventSource = SourceUtils.getEventSource(executionEnvironment);
        FlinkJedisPoolConfig redisConfig = new FlinkJedisPoolConfig.Builder().setHost("docker_server").build();
        eventSource.addSink(new RedisSink<>(redisConfig, new RedisMapper<Event>() {
            /**
             * 通过什么方式来操作redis
             */
            @Override
            public RedisCommandDescription getCommandDescription() {
                //用户的点击事件表
                return new RedisCommandDescription(RedisCommand.HSET, "users");
            }

            @Override
            public String getKeyFromData(Event event) {
                return event.user;
            }

            @Override
            public String getValueFromData(Event event) {
                return event.toString();
            }
        }));

        executionEnvironment.execute();
    }
}
